package com.grakol.parallel.samples;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Starts multiple RunnableTask objects in parallel threads. The actual 
 * start time of all the tasks is synchronized by using CyclicBarrier. 
 * Usage:
 * <ol>
 * <li>start() - starts runnable tasks in parallel threads;</li>
 * <li>getResults() - blocks until all runnable tasks finished; 
 * returns list of results;</li>
 * <li>getMergedResult() - can be used instead of or in addition to getResults(); 
 * blocks until all runnable tasks finished; returns merged result, if the merging
 * is supported by the 1st specified function;</li>
 * <li>finish() - terminates running thread pool.</li>
 * </ol>
 * <b>Note:</b> the MultiTaskLauncher cannot be re-initialized or re-started.
 * 
 * @author LK
 *
 * @param <R> - result type.
 */
public class MultiTaskLauncher<R> {
	private IFunction<R> _firstFunction;
	private ArrayList<RunnableTask<R>> _runTasks = new ArrayList<RunnableTask<R>>();
	private ExecutorService _threadPool;
	private CyclicBarrier _barrier;
	private AtomicBoolean _started = new AtomicBoolean();
	private AtomicBoolean _completed = new AtomicBoolean();
	private AtomicBoolean _finished = new AtomicBoolean();
	private ArrayList<R> _results = new ArrayList<R>();

	/**
	 * Constructs MultiTaskLauncher object for a given array of functions that 
	 * need to be called in parallel threads.
	 * 
	 * @param functions the given array of functions.
	 */
	public MultiTaskLauncher( IFunction<R>[] functions ) {
		if( functions.length < 1 )
			throw new IllegalArgumentException( "empty array" );
		_firstFunction = functions[0];
		_threadPool = Executors.newFixedThreadPool( functions.length );
		_barrier = new CyclicBarrier( functions.length );
		for( IFunction<R> function : functions ) 
			_runTasks.add( new RunnableTask<R>( function, _barrier ) );
	}
	/**
	 * Starts runnable tasks in parallel threads by using thread pool. 
	 */
	public synchronized void start() {
		if( _started.get() || _completed.get() || _finished.get() )
			throw new IllegalStateException( "cannot restart" );
		for( RunnableTask<R> task : _runTasks ) 
			_threadPool.execute( task );
		_started.set( true );
	}
	/**
	 * Blocks until all runnable tasks finished. Returns list of results.
	 * 
	 * @return the list of results after all runnable tasks finished.
	 */
	public synchronized List<R> getResults() {
		if( ! _started.get() )
			throw new IllegalStateException( "not started" );
		if( ! _completed.get() ) {
			for( RunnableTask<R> runTask : _runTasks )
				_results.add( runTask.getResult() );
			_completed.set( true );
		}
		return _results;
	}
	/**
	 * Blocks until all runnable tasks finished. Returns merged results from 
	 * all runnable tasks, if the merging is supported by the 1st specified 
	 * function.
	 * 
	 * @return the merged results from  all runnable tasks.
	 */
	public synchronized R getMergedResult() {
		return _firstFunction.mergeResults( getResults() );
	}
	/**
	 * Terminates running thread pool. This method has to be called at the end.
	 */
	public synchronized void finish() {
		try {
			_threadPool.shutdown();
			_threadPool.awaitTermination( 60, TimeUnit.SECONDS );
		} catch( Exception e ) {}
		_finished.set( true );
	}
	
}
